package net.techfuser.orderstatusreport;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.*;
import java.util.List;

class OrderStatusMySqlSink extends RichSinkFunction<List<OrderStatus>> {

    private Connection connection = null;
    private PreparedStatement ps = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        String url = "jdbc:mysql://qnh-flink-data.mysql.rds.aliyuncs.com:3306/flink_data_output?useUnicode=true&characterEncoding=utf-8";
        String user = "qnh_flink";
        String passwd = "Md7fpbWSHpBa";
        Class.forName("com.mysql.jdbc.Driver");
        connection = DriverManager.getConnection(url, user, passwd);
    }

    public void upsert(OrderStatus row) throws SQLException {
        String upsert = "INSERT INTO order_status_summary(" +
                "ent_id,  region_code, region_name, channel_keyword, date, " +
                "confirm_order_num, confirm_sale_value, package_order_num, package_sale_value, complate_order_num, " +
                "complate_sale_value, cancel_order_num, cancel_sale_value, valid_order_num, valid_sale_value) " +
                "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE " +
                "confirm_order_num = confirm_order_num + ?, " +
                "confirm_sale_value = confirm_sale_value + ?, " +

                "package_order_num = package_order_num + ?, " +
                "package_sale_value = package_sale_value + ?, " +

                "complate_order_num = complate_order_num + ?, " +
                "complate_sale_value = complate_sale_value + ?, " +

                "cancel_order_num = cancel_order_num + ?, " +
                "cancel_sale_value = cancel_sale_value + ?, " +

                "valid_order_num = valid_order_num + ?, " +
                "valid_sale_value = valid_sale_value + ?, " +

                "last_modify_time = ?";

        ps = connection.prepareStatement(upsert);
        ps.setLong(1, row.ent_id);
        ps.setString(2, row.region_code);
        ps.setString(3, row.region_name);
        ps.setString(4, row.channel_keyword);
        ps.setDate(5, new Date(row.date.getTime()));

        ps.setInt(6, row.confirm_order_num);
        ps.setDouble(7, row.confirm_sale_value);
        ps.setInt(8, row.package_order_num);
        ps.setDouble(9, row.package_sale_value);
        ps.setInt(10, row.complate_order_num);
        ps.setDouble(11, row.complate_sale_value);
        ps.setInt(12, row.cancel_order_num);
        ps.setDouble(13, row.cancel_sale_value);
        ps.setInt(14, row.valid_order_num);
        ps.setDouble(15, row.valid_sale_value);

        ps.setInt(16, row.confirm_order_num);
        ps.setDouble(17, row.confirm_sale_value);
        ps.setInt(18, row.package_order_num);
        ps.setDouble(19, row.package_sale_value);
        ps.setInt(20, row.complate_order_num);
        ps.setDouble(21, row.complate_sale_value);
        ps.setInt(22, row.cancel_order_num);
        ps.setDouble(23, row.cancel_sale_value);
        ps.setInt(24, row.valid_order_num);
        ps.setDouble(25, row.valid_sale_value);
        ps.setTimestamp(26, new Timestamp(System.currentTimeMillis()));

        ps.execute();
    }

    @Override
    public void invoke(List<OrderStatus> rows, Context context) throws Exception {
        for(OrderStatus row : rows){
            upsert(row);
        }
    }

    @Override
    public void close() throws Exception {
        if(null != connection) connection.close();
        if(null != ps) ps.close();
    }

}